package com.burges.net.dataStream.windows.join

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/4 
  * 描述     会话窗口关联操作实例
  */
object SessionWindowJoin {
	def main(args: Array[String]): Unit = {
		val parameterTool = ParameterTool.fromArgs(args)

		val env = StreamExecutionEnvironment.getExecutionEnvironment
		//创建黑色元素数据集
		val blackStream: DataStream[(Int, Long)] = env.fromElements((2, 21L), (4, 1L), (5, 4L))
		//创建白色元素数据集
		val whiteStream: DataStream[(Int, Long)] = env.fromElements((2, 21L), (1, 1L), (3, 4L))
		// 通过Join方法将两个数据集进行关联
		val windowStream = blackStream.join(whiteStream)
				.where(_._1)// 指定死一个Stream关联的key
				.equalTo(_._1)// 指定第二个Stream关联的key
				.window(EventTimeSessionWindows.withGap(Time.milliseconds(10)))// 指定窗口类型
				.apply((black, white) => (black._1, black._2 + white._2))// 应用JoinFunction
	}
}
